package streamforce.crypto.user;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Properties;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import streamforce.Tools;
import streamforce.crypto.KPAbe;
import streamforce.crypto.event.TransformedCTEventAll;
import streamforce.crypto.event.TransformedCTEventJoin;
import streamforce.crypto.event.UserInitEvent;
import streamforce.crypto.message.CloseConnection;
import streamforce.crypto.message.PolicyMessage;
import streamforce.crypto.message.UserHandshakeMessage;
import streamforce.crypto.policies.Policy;
import streamforce.crypto.user.subscriber.UserJoinPrice;
import streamforce.crypto.user.subscriber.UserJoinVolume;
import streamforce.crypto.user.subscriber.UserSelectAllSimple;
import streamforce.crypto.user.subscriber.UserSelectPriceSimple;

/**
 * Connecting to the server.
 * 
 * It only has to listen to new messages: initilization and
 * then decryption
 *
 */
public class EPUserClientJoin extends Thread{
	private final Log log = LogFactory.getLog(EPUserClientJoin.class); 
	
	static final boolean DEBUG = false; 
	
	public final static int SELECT_PRICE=0, SELECT_VOLUME=1, SELECT_ALL=2;
	ObjectInputStream ois=null; 
	ObjectOutputStream oos = null;
	Policy userPolicy; 
	int[] userIds, streamIds; 
	int type; 
	UserHandshakeMessage uhm; 
	boolean isDoneInit; 
	byte[][] pubK, tK, detK, decK; 
	String[] selectConditions; 
	KPAbe cryptoHandler; 
	int totalInitTime = 0; 
	public EPUserClientJoin(int uid1, int uid2, int type){		
		this.type = type;
		this.initialization();
		int[] ids = {uid1, uid2}; 
		this.uhm = new UserHandshakeMessage(ids,
				PolicyMessage.JOIN_QUERY);
		this.isDoneInit = true;
		this.cryptoHandler = new KPAbe();
	}
	
	public void initialization(){
		Properties property = new Properties();
		try {
			property.load(new FileInputStream("attributes.properties"));
			int port = new Integer(property.getProperty("serverPort")).intValue() + 1;
			Socket clientSocket = new Socket(property.getProperty("serverHost"), port);
			this.oos = new ObjectOutputStream(clientSocket.getOutputStream());
			this.ois = new ObjectInputStream(clientSocket.getInputStream());
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	//1. sent init message 
	//2. then listen to the TransformedCtEventAll event
	public void run(){
		try {
			this.oos.writeObject(this.uhm);						
			do{
				Object o = this.ois.readObject();
				if (o instanceof UserInitEvent){
					this.initializeUser((UserInitEvent)o); 
				}
				else if (o instanceof TransformedCTEventJoin){						
					this.userPolicy.update((TransformedCTEventJoin)o);
				}
				
			}while (true);
		} catch (Exception e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}				
	}
	
	private void initializeUser(UserInitEvent pmu){					
		if (this.type==EPUserClientJoin.SELECT_PRICE)
			this.userPolicy = new UserJoinVolume(this.cryptoHandler);
		else if (this.type == EPUserClientJoin.SELECT_VOLUME)
			this.userPolicy = new UserJoinPrice(this.cryptoHandler);
		long start = System.currentTimeMillis(); 
		this.userPolicy.initUser_simple(pmu.getUserId(), 2, pmu.getPubK()[0],
				pmu.gettK()[0], pmu.getDecK()[0], pmu.getDetK()[0]);
		long end = System.currentTimeMillis();
		this.totalInitTime += (end-start); 		
		this.isDoneInit = !this.isDoneInit;
				
		if (this.isDoneInit){//done init both users
			start = System.currentTimeMillis(); 
			this.pubK[1] = pmu.getPubK()[0];
			this.tK[1] = pmu.gettK()[0];
			this.decK[1] = pmu.getDecK()[0];
			//this.detK[1] = pmu.getDetK()[0];
			this.streamIds[1] = pmu.getStreamId();
			this.userIds[1] = pmu.getUserId();
			this.selectConditions[1] = pmu.getSelectCondition(); 
			
			byte[] joinKeys = this.userPolicy.getJoiningKey(this.userIds[0], this.userIds[1]);
			this.detK = new byte[2][joinKeys.length/2];
			System.arraycopy(joinKeys,0,this.detK[0],0,joinKeys.length/2);
			System.arraycopy(joinKeys,joinKeys.length/2,this.detK[1],0,joinKeys.length/2);
			
			String expj = "select s1.timestamp, s1.realTime, s1.stockId, s2.timestamp, s2.realTime, s2.stockId, " +
					" s1.price('ws=1'), s1.volume('ws=1'), s2.volume('ws=1')"
					+ " from StockEventJoin("
					+ this.selectConditions[0]
					+ ").win:length(8) as s1, StockEventJoin("
					+ this.selectConditions[1]
					+ ").win:length(8) as s2"
					+ " where s1.price('price_det') = s2.price('price_det')";
			
			PolicyMessage pm = new PolicyMessage(this.streamIds, this.userIds,
					this.pubK, this.tK, this.detK, this.decK, null, 0,
					PolicyMessage.SELECT_VOLUME, PolicyMessage.JOIN_QUERY, expj); 
			
			end = System.currentTimeMillis(); 
			this.totalInitTime +=(end-start); 
			log.info("user " + this.userIds[0] + " and " + this.userIds[1]
					+ " initialzation time = " + totalInitTime);
			try {
				this.oos.writeObject(pm);
				
				if (DEBUG)
					this.oos.writeObject(new CloseConnection()); 
				
				this.oos.flush();
				
				if (DEBUG)
					System.exit(0); 
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} 			
		}
		else{
			this.pubK = new byte[2][pmu.getPubK()[0].length];
			this.pubK[0] = pmu.getPubK()[0];
			this.tK = new byte[2][pmu.gettK()[0].length];
			this.tK[0] = pmu.gettK()[0];
			this.decK = new byte[2][pmu.getDecK()[0].length];
			this.decK[0] = pmu.getDecK()[0];
			
			//this.detK = new byte[2][pmu.getDetK()[0].length];
			//this.detK[0] = pmu.getDetK()[0];
			this.streamIds = new int[2]; 
			this.userIds = new int[2];
			this.userIds[0] = pmu.getUserId();
			this.streamIds[0] = pmu.getStreamId();
			this.selectConditions = new String[2]; 
			this.selectConditions[0] = pmu.getSelectCondition(); 
		}		
	}
		
	public static void main(String[] args){
		int id1 = new Integer(args[0]).intValue();
		int id2 = new Integer(args[1]).intValue();
		(new EPUserClientJoin(id1,id2,EPUserClientJoin.SELECT_VOLUME)).start();
	}
}
